package com.basker.pisces.redis.sequence;

import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import com.basker.pisces.core.sequence.ILongSequrenceGenerator;
import com.basker.pisces.redis.utils.RedisUtils;

public class RedisSequenceGenerator implements ILongSequrenceGenerator, InitializingBean {

    private static final String PREFIX = "seq";
    private ConcurrentHashMap<String, RedisSequence> sequences = new ConcurrentHashMap<>(256);

    @Autowired
    private RedisConnectionFactory factory;

    private RedisTemplate<String, Long> template;

    @Override
    public void afterPropertiesSet() throws Exception {
        template = new RedisTemplate<>();

        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericToStringSerializer<>(Long.class));
        template.setConnectionFactory(factory);

        template.afterPropertiesSet();
    }

    public long current(String sequenceName) {
        RedisSequence seq = this.getSequence(sequenceName);
        return seq.get();
    }

    @Override
    public long generate(String sequenceName) {
        RedisSequence seq = this.getSequence(sequenceName);
        return seq.incrementAndGet();
    }

    public void set(String sequenceName, Long value) {
        RedisSequence seq = this.getSequence(sequenceName);
        seq.set(value);
    }

    private RedisSequence getSequence(String sequenceName) {
        RedisSequence seq = sequences.get(sequenceName);
        if (seq == null) {
            String key = RedisUtils.resolveKeyWithApplicationName(PREFIX + ":" + sequenceName);
            seq = new RedisSequence(key, template);

            sequences.putIfAbsent(sequenceName, seq);
        }

        return seq;
    }

}
